package org.fuys.etp.registry.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.fuys.etp.core.EtpPersistentConfig;
import org.fuys.etp.core.EtpRegistryConfig;
import org.fuys.etp.core.model.res.EtpGlobalUpdateRes;
import org.fuys.etp.registry.IEtpRegistry;
import org.fuys.etp.registry.model.EtpRegistryHandleRes;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @projectName: fuys-etp-final
 * @package: org.fuys.etp.registry.impl
 * @className: ZookeeperEtpRegistry
 * @author: WangJing
 * @description: TODO
 * @date: 2025/3/1 12:10
 * @version: 1.0
 */
@Component
@Configuration
@Slf4j
public class ZookeeperEtpRegistry implements IEtpRegistry {

    @Resource
    private EtpPersistentConfig etpPersistentConfig;

    @Resource
    private EtpRegistryConfig etpRegistryConfig;

    private static final String BASE_CONFIG_PATH = "/fuys/etp/";

    private static final String HEALTH = "health";

    private static final String CONFIG = "config";

    @Resource(name = "fuysEtpZookeeper")
    private CuratorFramework zkClient;

    private final Map<String,CuratorCache> nodeCacheMap = new ConcurrentHashMap<>();

    @Override
    public EtpRegistryHandleRes registry() {
        String globalConfig = BASE_CONFIG_PATH + CONFIG + "/";
        String healthNode = BASE_CONFIG_PATH + HEALTH;
        try {
            etpPersistentConfig.toMap().forEach((k, v)->{
                try {
                    final String nodeKey = globalConfig + k;
                    zkClient.create().creatingParentsIfNeeded()
                            .forPath(nodeKey);
                    final CuratorCache build = CuratorCache.builder(zkClient, nodeKey).build();
                    nodeCacheMap.put(nodeKey,build);
                    build.start();
                    build.listenable().addListener((type,oldData,data)->{
                        if(CuratorCacheListener.Type.NODE_CHANGED.equals(type)){
                            final String[] split = data.getPath().split("/");
                            String configKey = split[split.length - 1];
                            final String value = new String(data.getData(), StandardCharsets.UTF_8);
                            final EtpGlobalUpdateRes update = etpPersistentConfig.updateByUpdater(configKey, value);
                            if(update.isResult()){
                                log.info("Fuys etp 全局配置更新成功。{}",configKey);
                            }else{
                                log.info("Fuys etp 全局配置更新失败。{}",configKey);
                            }
                            this.configListenerHandle(update.isResult()
                                    ,update.getMsg(),configKey,value,new String(oldData.getData(),StandardCharsets.UTF_8));
                        }
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            nodeCacheMap.put(healthNode,CuratorCache.builder(zkClient,healthNode).build());
        } catch (Exception e) {
            e.printStackTrace();
            return EtpRegistryHandleRes.builder().result(false).msg(e.getMessage()).build();
        }
        return EtpRegistryHandleRes.builder().result(true).build();
    }

    @Override
    public EtpRegistryHandleRes cancelRegistry() {
        nodeCacheMap.forEach((k,v)->{
            v.close();
        });
        try {
            zkClient.delete().deletingChildrenIfNeeded().forPath(BASE_CONFIG_PATH);
        } catch (Exception e) {
            e.printStackTrace();
            return EtpRegistryHandleRes.builder().result(false).msg(e.getMessage()).build();
        }
        return EtpRegistryHandleRes.builder().result(true).build();
    }

    @Override
    public EtpRegistryHandleRes configListenerHandle(boolean result, String msg, String key, Object value,
                                                     Object oldValue) {
        return null;
    }


    @Bean(name = "fuysEtpZookeeper")
    public CuratorFramework createZkConnectWithOptions(){
        ExponentialBackoffRetry backoffRetry = new ExponentialBackoffRetry(etpRegistryConfig.getRetryInterval(),
                etpRegistryConfig.getRetryAttempts());
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString(etpRegistryConfig.getAddress() + ":" + etpRegistryConfig.getPort())
                .retryPolicy(backoffRetry)
                .sessionTimeoutMs(etpRegistryConfig.getSessionTimeout())
                .connectionTimeoutMs(etpRegistryConfig.getConnectTimeout())
                .build();
        client.start();
        return client;
    }

    @PreDestroy
    public void destroy(){
        cancelRegistry();
        zkClient.close();
    }
}
